package net.bwie.realtime.jtp.dwd.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/06/12 10:00:19
 * @version: 1.0
 */
public class ReceiveKfKToDoris {
    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv();

        // 2. 输入表-input：映射到Kafka消息队列
        createInputTable(tabEnv);

        // 3. 数据处理-select
        Table resultTable = handle(tabEnv);

        // 4. 输出表-output：映射到Kafka队列
        createOutputTable(tabEnv);

        // 5. 保存数据-insert
        saveToKafka(tabEnv, resultTable);
    }

    private static void saveToKafka(TableEnvironment tabEnv, Table resultTable) {
        // 1. 将Table注册为表
        tabEnv.createTemporaryView("report_table", resultTable);

        // 2. 插入数据
        tabEnv.executeSql(
                "INSERT INTO dws_trade_province_order_report_doris_sink\n" +
                        "SELECT\n" +
                        "    SUBSTRING(CAST(start_time AS STRING), 0, 19) AS start_time,\n" +
                        "    SUBSTRING(CAST(end_time AS STRING), 0, 19) AS end_time,\n" +
                        "    SUBSTRING(CAST(start_time AS STRING), 0, 10) AS cur_date,\n" +
                        "    province_id, province_name, region_id, region_name,\n" +
                        "    order_count, order_amount, activity_amount, coupon_amount\n" +
                        "FROM report_table"
        );
    }


    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql(
                "CREATE TABLE dws_trade_province_order_report_doris_sink(\n" +
                        "    `start_time` STRING,\n" +
                        "    `end_time` STRING,\n" +
                        "    `cur_date` STRING,\n" +
                        "    `province_id` STRING,\n" +
                        "    `province_name` STRING,\n" +
                        "    `region_id` STRING,\n" +
                        "    `region_name` STRING,\n" +
                        "    `order_count` BIGINT,\n" +
                        "    `order_amount` DECIMAL(16, 2),\n" +
                        "    `activity_amount` DECIMAL(16, 2),\n" +
                        "    `coupon_amount` DECIMAL(16, 2)\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'jtp_mall_report.dws_trade_province_order_report',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.batch.interval' = '10s',\n" +
                        "    'sink.max-retries' = '3',\n" +
                        "    'sink.batch.size' = '1000'\n" +
                        ")"
        );
    }


    private static Table handle(TableEnvironment tabEnv) {
        // s1-设置窗口，省份分组，聚合计算
        Table resultTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') AS start_time\n" +
                        "    , DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') AS end_time\n" +
                        "    , province_id\n" +
                        "    , count(DISTINCT order_id) AS order_count\n" +
                        "    , sum(cast(split_total_amount AS DECIMAL(10, 2))) AS order_amount\n" +
                        "    , sum(cast(if(split_activity_amount IS NULL, '0', split_activity_amount) AS DECIMAL(10, 2)))  AS activity_amount\n" +
                        "    , sum(cast(if(split_coupon_amount IS NULL, '0', split_coupon_amount) AS DECIMAL(10, 2)))  AS coupon_amount\n" +
                        "    , PROCTIME() AS proc_time\n" +
                        "FROM TABLE (\n" +
                        "    TUMBLE(TABLE dwd_order_detail_kafka_source, DESCRIPTOR(create_time), INTERVAL '1' MINUTES)\n" +
                        ")\n" +
                        "WHERE user_id IS NOT NULL AND source_type_name IS NOT NULL\n" +
                        "GROUP BY window_start, window_end, province_id"
        );
//resultTable.execute().print();
        tabEnv.createTemporaryView("result_table", resultTable);

// s2-维度表：dim_base_province
        tabEnv.executeSql(
                "CREATE TABLE dim_base_province_hbase_source (\n" +
                        "    row_key STRING,\n" +
                        "    info ROW<area_code STRING, id STRING, iso_3166_2 STRING, iso_code STRING, name STRING, region_id STRING>,\n" +
                        "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'hbase-2.2',\n" +
                        "    'table-name' = 'dim_base_province',\n" +
                        "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                        "    'lookup.async' = 'true',\n" +
                        "    'lookup.cache.max-rows' = '34',\n" +
                        "    'lookup.cache.ttl' = '1 hour'\n" +
                        ")"
        );

        tabEnv.executeSql(
                "CREATE TABLE dim_base_region_hbase_source (\n" +
                        "    row_key STRING,\n" +
                        "    info ROW<id STRING, region_name STRING>,\n" +
                        "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'hbase-2.2',\n" +
                        "    'table-name' = 'dim_base_region',\n" +
                        "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                        "    'lookup.async' = 'true',\n" +
                        "    'lookup.cache.max-rows' = '34',\n" +
                        "    'lookup.cache.ttl' = '1 hour'\n" +
                        ")"
        );
// tabEnv.executeSql("SELECT * FROM dim_base_region_hbase_source").print();

// s3-关联省份维表数据
        Table reportTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    t1.start_time,\n" +
                        "    t1.end_time,\n" +
                        "    t1.province_id,\n" +
                        "    t2.name AS province_name,\n" +
                        "    t2.region_id,\n" +
                        "    t3.region_name,\n" +
                        "    t1.order_count,\n" +
                        "    t1.order_amount,\n" +
                        "    t1.activity_amount,\n" +
                        "    t1.coupon_amount,\n" +
                        "    UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        "FROM result_table t1\n" +
                        "    LEFT JOIN dim_base_province_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                        "       ON t1.province_id = t2.row_key\n" +
                        "    LEFT JOIN dim_base_region_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t3\n" +
                        "       ON t2.region_id = t3.row_key"
        );
        //reportTable.execute().print();

        // 返回结果
        return reportTable ;
    }

    private static void createInputTable(TableEnvironment tabEnv) {

        tabEnv.executeSql("");

        tabEnv.executeSql(
                "CREATE TABLE dwd_order_detail_kafka_source(\n" +
                        "    `id` STRING,\n" +
                        "    `order_id` STRING,\n" +
                        "    `user_id` STRING,\n" +
                        "    `order_status` STRING,\n" +
                        "    `sku_id` STRING,\n" +
                        "    `sku_name` STRING,\n" +
                        "    `province_id` STRING,\n" +
                        "    `activity_id` STRING,\n" +
                        "    `activity_rule_id` STRING,\n" +
                        "    `coupon_id` STRING,\n" +
                        "    `date_id` STRING,\n" +
                        "    `create_time` TIMESTAMP(3),\n" +
                        "    `operate_date_id` STRING,\n" +
                        "    `operate_time` STRING,\n" +
                        "    `source_id` STRING,\n" +
                        "    `source_type` STRING,\n" +
                        "    `source_type_name` STRING,\n" +
                        "    `sku_num` STRING,\n" +
                        "    `split_original_amount` STRING,\n" +
                        "    `split_activity_amount` STRING,\n" +
                        "    `split_coupon_amount` STRING,\n" +
                        "    `split_total_amount` STRING,\n" +
                        "    `row_op_ts` STRING,\n" +
                        "    `proc_time` AS PROCTIME(),\n" +
                        "    WATERMARK FOR create_time AS create_time - INTERVAL '0' MINUTE\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dwd-order-detail',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "    'properties.group.id' = 'gid-dws-trade-sku-order',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json',\n" +
                        "    'json.fail-on-missing-field' = 'false',\n" +
                        "    'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );
    }


    private static TableEnvironment getTableEnv() {
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // 3返回对象
        return tabEnv;
    }
}
